package com.etlmaster.executor.scheduler.utils;

import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.util.HashMap;
import java.util.Map;

import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.ZooKeeper;
import com.etlmaster.executor.scheduler.ScheduleServer;
import com.etlmaster.executor.utils.LogWriter;

/**
 * @author Larry
 * @copyright Larry@etlMaster
 * @since 2017-09-10
 */
public class ZkUtils {
	private static ZooKeeper zk;

	public static ZooKeeper getInstance(String host, int sessiontimeout,
			Watcher wh) throws IOException {
		if (zk == null)
			zk = new ZooKeeper(host, sessiontimeout, wh);
		return zk;

	}

	public static ZooKeeper getInstance() {
		return zk;
	}

	private ZkUtils() {

	}

	private static Watcher wh;

	public static Watcher getWatcher() {
		if (wh == null)
			wh = new Watcher() {

				public void process(WatchedEvent event) {

					try {
						if (event.getPath() != null) {
							String data = new String(zk.getData(
									event.getPath(), true, null), "utf-8");
							LogWriter.addLog("INFO",
									"回调watcher实例： 路径" + event.getPath()
											+ " 类型："

											+ event.getType() + " 数据 " + data);
							Map<String, String> eventMap = new HashMap<String, String>();
							eventMap.put("path", event.getPath());
							eventMap.put("type", event.getType().toString());
							eventMap.put("data", data);
							ScheduleServer.getEventLock().lock();
							ScheduleServer.getEventMaps().add(eventMap);
							ScheduleServer.getNotEmpty().signal();
							ScheduleServer.getEventLock().unlock();

						}
					} catch (KeeperException e) {
						// TODO Auto-generated catch block
						e.printStackTrace();
					} catch (InterruptedException e) {
						// TODO Auto-generated catch block
						e.printStackTrace();
					} catch (UnsupportedEncodingException e) {
						// TODO Auto-generated catch block
						e.printStackTrace();
					}

				}

			};
		return wh;
	}

	public String createNode(String path, byte[] data) throws KeeperException,
			InterruptedException {
		return zk
				.create(path, data, Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
	}

}
